package mysql;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.types.Row;

public class MysqlDataSet
{
    public static void main(String[] args) throws Exception
    {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<Row> dbData =
        env.createInput(
                JDBCInputFormat.buildJDBCInputFormat()
                        .setDrivername(ConfigKeys.DRIVER_CLASS)
                        .setDBUrl(ConfigKeys.SOURCE_DRIVER_URL)
                        .setUsername(ConfigKeys.SOURCE_USER)
                        .setPassword(ConfigKeys.SOURCE_PASSWORD)
                        .setQuery("SELECT goods_id,goods_name FROM goods_info")
                        .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO))
                        .finish()
        );

        dbData.output(
                JDBCOutputFormat.buildJDBCOutputFormat()
                        .setDrivername(ConfigKeys.DRIVER_CLASS)
                        .setDBUrl(ConfigKeys.SOURCE_DRIVER_URL)
                        .setUsername(ConfigKeys.SOURCE_USER)
                        .setPassword(ConfigKeys.SOURCE_PASSWORD)
                        .setQuery("INSERT INTO goods_info_sink(goods_id,goods_name) values(?,?)")
                        .finish()
        );

        env.execute("flink mysql Dataset demo");
    }
}
